package com.burges.net.dataStream.codeRuler.transform.MultiDataStream

import org.apache.flink.streaming.api.scala.{ConnectedStreams, DataStream, StreamExecutionEnvironment, _}

/**
  * 创建人    BurgessLee
  * 创建时间   2020/1/26
  * 描述     不太明了
  */
object IterateDemo {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		//为了对数据分区根据默认的并行度进行重平衡
		val dataStream = environment.fromElements(3,1,2,1,5).map{t: Int => t}

		/**
		  * 实现的目的是如果事件指标+1后等于2，那么计算指标反馈到下一次迭代的同道中人，如果时间指标+1不等于2，那么直接输出到下游的DataStream中
		  */
		val iterateStream: DataStream[String] = dataStream.iterate((input: ConnectedStreams[Int, String]) => {
			//分别定义两个map方法完成对输入的数据集的数据处理
			val head = input.map(i => (i + 1).toString, s => s)
			(head.filter(_ == "2"), head.filter(_ != "2"))
			//1000指定最长迭代时间，超过改时间没有数据接入就终止迭代，单位ms
		}, 1000)
		iterateStream.print()
		environment.execute("streaming iterate operator")

	}

}
